package com.shujia.flink.core

import org.apache.flink.streaming.api.scala._

object Demo1WordCount {
  def main(args: Array[String]): Unit = {

    /**
      * 创建flink的环境
      *
      * getExecutionEnvironment: 获取flink的环境，如果有直接返回，如果没有再创建
      *
      */

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * 设置默认的并行度
      *
      */
    //env.setParallelism(2)


    /**
      * 1、读取数据
      * nc -lk 8888
      */

    val lineDS: DataStream[String] = env.socketTextStream("master", 8888)


    /**
      * 处理数据
      * 统计单词的数量
      *
      */
    val wordsDS: DataStream[String] = lineDS.flatMap(_.split(","))


    /**
      * 转换成kv格式
      *
      */

    val kvDS: DataStream[(String, Int)] = wordsDS.map((_, 1))


    /**
      * 安装单词分组
      *
      */
    val keyByDS: KeyedStream[(String, Int), String] = kvDS.keyBy(_._1)


    /**
      * 分组之后进行聚合
      *
      */

    val countDS: DataStream[(String, Int)] = keyByDS.reduce((x, y) => (x._1, x._2 + y._2))


    /**
      * 打印结果
      *
      */
    countDS.print()

    /**
      * 启动flink程序
      *
      */

    env.execute()


  }


}
